package com.verlet.springbootdemo.simple.rabbitmq;

import com.verlet.springbootdemo.simple.rabbitmq.converter.TextMessageConverter;
import com.verlet.springbootdemo.simple.rabbitmq.delegate.MessageDelegate;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.UUID;

/**
 * 声明交换机和队列
 *
 * @author verlet
 * @date 2020/6/27
 */
@Configuration
@ConditionalOnProperty(prefix = "rabbitmq", name = "enabled", havingValue = "true", matchIfMissing = true)
public class RabbitmqConfig {

    public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    public static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    public static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
    public static final String ROUTINGKEY_EMAIL = "inform.#.email.#";
    public static final String ROUTINGKEY_SMS = "inform.#.sms.#";

    /**
     * 声明TOPICS工作模式的交换机
     */
    @Bean(EXCHANGE_TOPICS_INFORM)
    public Exchange exchangeTopicsInform() {
        // durable(true) 表面重启之后交换机还在
        return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build();
    }

    /**
     * 声明QUEUE_INFORM_EMAIL队列
     */
    @Bean(QUEUE_INFORM_EMAIL)
    public Queue queueInformEmail() {
        return QueueBuilder.durable(QUEUE_INFORM_EMAIL).build();
    }

    /**
     * 声明QUEUE_INFORM_SMS队列
     */
    @Bean(QUEUE_INFORM_SMS)
    public Queue queueInformSms() {
        return QueueBuilder.durable(QUEUE_INFORM_SMS).build();
    }

    /**
     * 交换机与QUEUE_INFORM_EMAIL队列绑定
     */
    @Bean
    public Binding bindingQueueInformEmail(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue,
                                           @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_EMAIL).noargs();
    }

    /**
     * 交换机与QUEUE_INFORM_SMS队列绑定
     */
    @Bean
    public Binding bindingQueueInformSms(@Qualifier(QUEUE_INFORM_SMS) Queue queue,
                                         @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_SMS).noargs();
    }

    /**
     * spring 消费监听容器
     */
    @Bean
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer messageContainer = new SimpleMessageListenerContainer(connectionFactory);
        // 监控多个队列
        messageContainer.setQueues(queueInformEmail(), queueInformSms());
        // 当前消费者数量
        messageContainer.setConcurrentConsumers(1);
        // 最大消费者数量
        messageContainer.setMaxConcurrentConsumers(5);
        // 重回队列
        messageContainer.setDefaultRequeueRejected(false);
        // ACK签收模式
        messageContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
        // Tag策略
        messageContainer.setConsumerTagStrategy(queue -> queue + "_" + UUID.randomUUID().toString());

        // 添加监听器
       /* messageContainer.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                System.out.println(new String(message.getBody()));
            }
        });*/

        // 使用消息监听适配器
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(new MessageDelegate());
        messageListenerAdapter.setDefaultListenerMethod("consumeMessage");
        // queue和方法名映射
//        messageListenerAdapter.setQueueOrTagToMethodName();

        /**
         ContentTypeDelegatingMessageConverter = new ContentTypeDelegatingMessageConverter();

         TextMessageConverter textMessageConverter = new TextMessageConverter();
         contentTypeDelegatingMessageConverter.addDelegate(MessageProperties.CONTENT_TYPE_TEXT_PLAIN, textMessageConverter);
         contentTypeDelegatingMessageConverter.addDelegate(MessageProperties.CONTENT_TYPE_XML, textMessageConverter);

         Json转Java对象
         Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
         DefaultJackson2JavaTypeMapper defaultJackson2JavaTypeMapper = new DefaultJackson2JavaTypeMapper();
         Map<String, Class<?>> idClassMapping = new HashMap();
         idClassMapping.put("messageDelegate", MessageDelegate.class);
         defaultJackson2JavaTypeMapper.setIdClassMapping(idClassMapping);
         jackson2JsonMessageConverter.setJavaTypeMapper(defaultJackson2JavaTypeMapper);
         contentTypeDelegatingMessageConverter.addDelegate(MessageProperties.CONTENT_TYPE_JSON,jackson2JsonMessageConverter);
         */

        // 设置Message转换器
        messageListenerAdapter.setMessageConverter(new TextMessageConverter());
        messageContainer.setMessageListener(messageListenerAdapter);
        return messageContainer;
    }

}